package scala.actors.multicast.actorgroup

import edu.ucla.cs.rpc.multicast.Sender
import edu.ucla.cs.rpc.multicast.network.MulticastManager
import edu.ucla.cs.rpc.multicast.sequencer.SequencerDest
import edu.ucla.cs.rpc.multicast.sequencer.SequencerSender

import scala.actors.multicast.handlers.ActorGroupMessageHandler
import scala.actors.Actor.actor
import scala.actors.Actor.loop
import scala.actors.Actor.self
import scala.actors.multicast.message.ActorCommand
import scala.actors.multicast.message.ActorMessage
import scala.actors.multicast.message.Bang
import scala.actors.multicast.message.BangMark
import scala.actors.multicast.message.BangBang
import scala.actors.multicast.message.Forward
import scala.collection.mutable.HashSet

import java.io.Serializable
import java.net.SocketAddress

/*
  class MulticastActorGroup
  @author Chase Covello
  @author Micah Wendell
  @author Philip Russell
  
  Conceptually, a MulticastActorGroup serves as a relay between the MulticastRPC
  framework. It has two communication portals: Sender (outgoing) and Dest (incoming).
  It processes/forwards messages between these two portals and the local group of actors
  running in the virtual machine.
  */
/*
  The set of exceptions specific to a MulticastActorGroup. These are thrown in case
  of invalid input arguments to the primary constructor ie if they are NULL.
  */
abstract class ActorGroupException extends Exception
case object SequencerAddressIsNull extends ActorGroupException
case object ManagerAddressIsNull extends ActorGroupException

/*
  A MulticastActorGroup is a relay that forwards messages FROM a local actor TO a Sender object,
  and forwards messages FROM a Dest object TO the members of its local group.
  Inputs: sequencers, a Collection of SocketAddress(es) belonging to the Sequencers that the
  Sender communicates with; manager, the SocketAddress of the MulticastManager that facilitates
  communication in the MulticastRPC framework.
  */
class MulticastActorGroup(sequencers: Collection[SocketAddress], manager: SocketAddress) extends ActorGroup {

    //Check that the inputs are non-NULL, and throw relevant exception if they are
  if (sequencers eq null)
    throw SequencerAddressIsNull
  else if (manager eq null)
    throw ManagerAddressIsNull

    //Initialize mcastSender (outgoing communication portal) with the given Sequencer addresses
    /*Initialize mcastDest (incoming communication portal) with the given MulticastManager
      address and a MessageHandler object that will pass messages FROM the Dest object TO
      this MulticastActorGroup*/
  private val mcastSender: Sender = new SequencerSender(sequencers.toArray[java.net.SocketAddress])
  private val mcastDest: SequencerDest = new SequencerDest(manager, new ActorGroupMessageHandler(this))
  private var idString: String = "ACTORGROUP"

    //Alternate constructor with a String parameter to identity the MulticastActorGroup
    //Primarily useful for debugging/println statements
  def this(name: String, seqs: Collection[SocketAddress], manager: SocketAddress) = {
    this(seqs, manager)
    idString = name
  }
  
    /*The main logic loop of a MulticastActorGroup. The group continues awaiting and responding
      to messages until it receives a Shutdown message*/
  override def act: Unit = {
    loop {
      receive {
        
        /* Pattern match for an incoming Pair that includes the Bang (!)
           command. For each actor in the group, asynchronously deliver the
           message by invoking its ! method. */
        case ActorMessage(Bang, msg, _) => actors.foreach(actor => if (actor != sender) actor ! msg)
    
        /* Same as the above, but for the scala.actors.Actor forward method. */
        case ActorMessage(Forward, msg, _) => actors.foreach(actor => if (actor != sender) actor forward msg)
    
        /* BangMark corresponds to scala.actors.Actor.!?, synchronous message
           sending and waiting for a reply. We declare a Collection(HashSet)
           of messages, then synchronously spawn a new Actor that will invoke
           the !? method for a particular Actor in the local set of Actors.
           
           NOTE: attempting to perform the core logic here in THIS actor will
           result in deadlock, because THIS actor is already synchronously
           waiting (due to the semantics of the scala.actors.Actor.receive
           method. */
        case ActorMessage(BangMark, msg, source) => {
          val responses = new HashSet[Any] //The set of responses to collect
          actor { //Spawn an actor
            actors -= self //Remove this actor from the set of actors
            responses.synchronized { //Synchronously invoke the !? method
              actors.foreach(actor => if (actor != sender) responses += (actor !? msg))
              responses.notifyAll //release the lock on the responses object
            }
          }
          responses.synchronized {
            responses.wait
            new ActorMessage(Bang, responses.asInstanceOf[Serializable], mcastDest.getSocketAddress).send(source)
          }
        }
    
        /* As above, but for Futures. */
        case ActorMessage(BangBang, msg, source) => {
          val responses = new HashSet[Any]
          actor {
            actors -= self
            responses.synchronized {
              actors.foreach(actor => if (actor != sender) responses += (actor !! msg))
              responses.notifyAll
            }
          }
          responses.synchronized {
            responses.wait
            new ActorMessage(Bang, responses.asInstanceOf[Serializable], mcastDest.getSocketAddress).send(source)
          }
        }
      
        /* Shutdown is a message based method of shutting down this
           MulticastActorGroup. */
        case Shutdown => {
          mcastDest.shutdown
          exit('shutdown)
        }
        /* Messages of this form are FROM a local ActorGroup. Refer to
           scala.actors.multicast.actorgroup.ActorGroup */
        case (cmd: ActorCommand, msg) => mcastSender.totalOrderBroadcast(new ActorMessage(cmd, msg.asInstanceOf[Serializable], mcastDest.getSocketAddress))
      }
    }
  }

  override def toString: String = idString + ": " + super.toString
}
